package com.example.KafkaDdemo.product;


import com.example.KafkaDdemo.uitls.BytesUtils;
import org.apache.kafka.clients.producer.ProducerRecord;
import org.apache.kafka.common.header.internals.RecordHeader;
import org.apache.kafka.common.header.internals.RecordHeaders;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.kafka.core.KafkaTemplate;
import org.springframework.kafka.support.SendResult;
import org.springframework.messaging.Message;
import org.springframework.web.bind.annotation.*;

import java.util.concurrent.ExecutionException;

@RestController
public class KafkaProducer {
    @Autowired
    private KafkaTemplate<String, Object> kafkaTemplate;

    // 发送消息
    @GetMapping("/kafka/sync/{message}")
    public void sendMessageSync(@PathVariable("message") String normalMessage) {

        for (int i = 0; i < 20; i++) {
            try {
                SendResult<String, Object> result = kafkaTemplate.send("topic1", normalMessage).get();
            } catch (Exception e) {
                e.printStackTrace();
            }
        }
    }

    @PostMapping("/kafka/callbackOne")
    public void sendMessage2() {
        for (int i = 0; i < 100; i++) {
            long l = System.currentTimeMillis();
            kafkaTemplate.send("vxtopic", String.valueOf(l), "消息" + i).addCallback(success -> {
                // 消息发送到的topic
                String topic = success.getRecordMetadata().topic();
                // 消息发送到的分区
                int partition = success.getRecordMetadata().partition();
                // 消息在分区内的offset
                long offset = success.getRecordMetadata().offset();
                System.out.println("发送消息成功:" + topic + "-" + partition + "-" + offset);
            }, failure -> {
                System.out.println("发送消息失败:" + failure.getMessage());
            });
        }
    }

    @PostMapping("/kafka/ttl")
    public void sendMessage3(@RequestParam(value = "message1") String message1,@RequestParam(value = "message2") String message2,
                             @RequestParam(value = "message3") String message3) throws ExecutionException, InterruptedException {
        ProducerRecord<String, Object> record1 = new ProducerRecord<>("ttL_topic", 1, System.currentTimeMillis(),
                null, message1);

        // 超时消息
        ProducerRecord<String, Object> record2 = new ProducerRecord<>("ttL_topic", 1, System.currentTimeMillis() - 5 * 1000,
                null, message2, new RecordHeaders().add(new RecordHeader("ttl", BytesUtils.longToBytes(5))));

        ProducerRecord<String, Object> record3 = new ProducerRecord<>("ttL_topic", 1, System.currentTimeMillis(),
                null, message3);

        kafkaTemplate.send(record1).get();
        kafkaTemplate.send(record3).get();
        kafkaTemplate.send(record2).get();

    }


}
